//package com.boot.canal.config;
//
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.boot.CommandLineRunner;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.Resource;
//
/**
 * @Author: lkz
 * @Title: CanalRunner
 * @Description: 项目启动时加载
#canal:
#  open: 2 #1 开启监听 2关闭
#  hostname: 127.0.0.1
#  destination: example
#  subscribe: sunda
 * @Date: 2023/9/18 13:50
 */
////@Component
//@Slf4j
//public class CanalRunner implements CommandLineRunner {
//
//
//
//    @Value("${canal.open}")
//    private String canalOpen;
//
//    @Value("${canal.hostname}")
//    private String canalHostname;
//
//    @Value("${jianfan.canal.port}")
//    private String canalPort;
//
//
//    @Value("${jianfan.canal.subscribe}")
//    private String canalSubscribe;
//
//
//    @Value("${jianfan.canal.destination}")
//    private String canalDestination;
//
//    @Override
//    public void run(String... args) throws Exception {
//
//        //
////        if (StrUtil.equals(canalOpen,"2")){
////            return;
////        }
////
////        // 创建一个 CanalConnector 连接器
////        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalHostname, Integer.valueOf(canalPort)), canalDestination, "", "");
////
////        try {
////            // 连接 Canal Server，尝试多次重连
////            while (true) {
////                try {
////                    canalConnector.connect();
////                    break;
////                } catch (CanalClientException e) {
////                    System.out.println("Connect to Canal Server failed, retrying...\n");
////                }
////            }
////            //订阅数据库表,默认监听所有的数据库、表，等同于：.*\\..*
////            canalConnector.subscribe(canalSubscribe);
////            // 回滚到上一次的 batchId，取消已经消费过的日志
////            canalConnector.rollback();
////
////            // 持续监听 Canal Server 推送的数据，并将数据写入 Redis 中
////            while (true) {
////                Message message = canalConnector.getWithoutAck(100);
////                long batchId = message.getId();
////
////                // 如果没有新数据，则暂停固定时间后继续获取
////                if (batchId == -1 || message.getEntries().isEmpty()) {
////                    try {
////                        Thread.sleep(1000);
////                    } catch (InterruptedException e) {
////                        e.printStackTrace();
////                    }
////                } else {
////                    for (CanalEntry.Entry entry : message.getEntries()) {
////
////                        if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
////                            continue;
////                        }
////
////                        CanalEntry.RowChange rowChange = null;
////                        try {
////                            rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
////                        } catch (Exception e) {
////                            throw new RuntimeException("Error parsing Canal Entry.", e);
////                        }
////
////                        String schema = entry.getHeader().getSchemaName();
////                        String table = entry.getHeader().getTableName();
////                        CanalEntry.EventType eventType = rowChange.getEventType();
////
////                        // 变更后的新数据
////                        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
////                           if (eventType == CanalEntry.EventType.INSERT) {
////                                //添加数据订阅
////                               insertData(schema, table, rowData);
////                           } else if (eventType == CanalEntry.EventType.UPDATE){
////                                //修改数据订阅
////                               updateData(schema, table, rowData);
////                           } else if (eventType == CanalEntry.EventType.DELETE){
////                               //删除数据
////                               deleteData(schema, table, rowData);
////                           }
////                        }
////                    }
////
////                    try {
////                        canalConnector.ack(batchId);
////                    } catch (Exception e) {
////                        // 回滚所有未确认的 Batch
////                        canalConnector.rollback(batchId);
////                    }
////                }
////            }
////        } finally {
////            canalConnector.disconnect();
////        }
////    }
//
//
//    }
//}
